package net.bgraham.collectionbatcher;

import java.util.Collection;
import java.util.HashSet;
import java.util.Set;

/**
 * This class provides support for collecting data until one of two events occur:
 * 1) The specified "maxTimeToWaitBetweenBatches" expires, or
 * 2) The specified "maxTimeToWaitBetweenUpdates" expires.
 * Once either of these events occurs, the listeners are called back with the data
 * that has been collected, the data structure is cleared, and the cycle is repeated
 * until the "exit()" method is called.
 * "maxTimeToWaitBetweenBatches" is defined as the maximum amount of time to wait
 * before calling back the listeners with the current data.
 * "maxTimeToWaitBetweenUpdates" is defined as the maximum amount of time to wait
 * between calls to the "add()" method before calling back the listeners with the
 * current data.
 * For example, if "maxTimeToWaitBetweenBatches" = 1000 milliseconds and
 * "maxTimeToWaitBetweenUpdates" = 100 milliseconds, the following scenarios are possible:
 * 1) No calls to "add()" data are ever made, so the listeners will never be called.
 * 2) A steady stream of calls to "add()" (with < 100 ms between calls) will result in a
 * callback to the listeners after 1000 ms. 
 * 3) A burst of calls to "add()" followed by > 100 ms before "add()" is called again
 * will result in the listeners being called back 100 ms after the last "add()".
 * Usage: Create a Batcher, call "add()" to add data to be collected/batched, and call
 * "exit()" to stop and dispose the Batcher.
 * See the unit tests and functional tests for example usage.
 * @param <T>
 */
public class Batcher<T>
{
	private final Set<IBatchListener<T>> batchListeners;
	private final Collection<T> batchData;
	private final BatchTimer batchTimer;
	
	/**
	 * Creates a Batcher that will collect data and call back the specified 
	 * listener with the collected data when one of two events occur:
	 * 1) The specified "maxTimeToWaitBetweenBatches" expires, or
	 * 2) The specified "maxTimeToWaitBetweenUpdates" expires.
	 * @param batchListener
	 * @param maxTimeToWaitBetweenBatches The maximum amount of time to wait before
	 * calling back the listeners with the current data.
	 * @param maxTimeToWaitBetweenUpdates The maximum amount of time to wait between
	 * calls to the "add()" method before calling back the listeners with the
	 * current data.
	 * @param batchData The Collection class instance to use to store the collected
	 * data, such as a List for chronological collection of all data or a Set for
	 * handling conflation of data to remove duplicates.
	 */
	public Batcher(final IBatchListener<T> batchListener, final long maxTimeToWaitBetweenBatches,
			final long maxTimeToWaitBetweenUpdates, final Collection<T> batchData)
	{
		batchListeners = new HashSet<IBatchListener<T>>();
		batchListeners.add(batchListener);
		this.batchData = batchData;
		
		batchTimer = new BatchTimer("batchTimer",
				new IBatchTimerListener()
			{
				public void cancelled()
				{
				}
	
				public void timedOut(boolean fullTimeoutHit)
				{
					sendBatchDataToListeners(fullTimeoutHit);
				}
			},
			maxTimeToWaitBetweenBatches,
			maxTimeToWaitBetweenUpdates);
	}

	private void sendBatchDataToListeners(boolean fullTimeoutHit)
	{
		synchronized (batchListeners)
		{
			synchronized (batchData)
			{
				if (batchData.size() > 0)
				{
					for (IBatchListener<T> listener : batchListeners)
					{
						listener.batchCollected(batchData, fullTimeoutHit);
					}
					batchData.clear();
				}
			}
		}
	}

	public void addBatchListener(IBatchListener<T> listener)
	{
		synchronized (batchListeners)
		{
			batchListeners.add(listener);
		}
	}
	
	public boolean removeBatchListener(IBatchListener<T> listener)
	{
		synchronized (batchListeners)
		{
			return batchListeners.remove(listener);
		}
	}
	
	public void removeAllBatchListeners()
	{
		synchronized (batchListeners)
		{
			for (IBatchListener<T> listener : batchListeners)
			{
				batchListeners.remove(listener);
			}
		}
	}
	
	private void addBatchData(T item)
	{
		synchronized (batchData)
		{
			batchData.add(item);
		}
	}
	
	/**
	 * Call "add(T item)" to add a data item to be collected/batched.
	 * @param item
	 */
	public void add(T item)
	{
		addBatchData(item);
		batchTimer.start();
	}
	
	/**
	 * Call "exit()" when you no longer wish to use the Batcher to have it shut
	 * down nicely.  This method will not return until the BatchTimer thread has been
	 * shut down and all listeners have been called back with any outstanding data.
	 */
	public void exit()
	{
		batchTimer.exit();
		batchTimer.join();
		sendBatchDataToListeners(false);
		removeAllBatchListeners();
	}
	
	private class BatchTimer implements Runnable
	{
		private final IBatchTimerListener batchTimerListener;
		private final long maxTimeToWaitBetweenBatches;
		private final long maxTimeToWaitBetweenUpdates;
		private boolean isTimerCurrentlyRunning = false;
		private boolean cancelTimer = false;
		private boolean exit = false;
		private final Thread thread;
		private long nextTimeToTimeoutBatch = 0;
		private long nextTimeToTimeoutUpdates = 0;
		
		BatchTimer(final String name, final IBatchTimerListener batchTimerListener,
				final long maxTimeToWaitBetweenBatches,
				final long maxTimeToWaitBetweenUpdates)
		{
			this.batchTimerListener = batchTimerListener;
			this.maxTimeToWaitBetweenBatches = maxTimeToWaitBetweenBatches;
			this.maxTimeToWaitBetweenUpdates = maxTimeToWaitBetweenUpdates;
			thread = new Thread(this, name);
			thread.start();
		}
		
		public synchronized void start()
		{
			nextTimeToTimeoutUpdates = System.currentTimeMillis() + maxTimeToWaitBetweenUpdates;
			
			if (isTimerCurrentlyRunning)
			{
				return;
			}
			
			nextTimeToTimeoutBatch = System.currentTimeMillis() + maxTimeToWaitBetweenBatches;
			notify();
		}
		
		public synchronized void cancel()
		{
			cancelTimer = true;
			notify();
		}
		
		public synchronized void exit()
		{
			exit = true;
			notify();
		}
		
		public void join()
		{
			try
			{
				thread.join();
			} catch (InterruptedException e)
			{
			}
		}

		public synchronized void run()
		{
			while (true)
			{
				try
				{
					isTimerCurrentlyRunning = false;
					wait();
				} catch (InterruptedException e1)
				{
				}
				isTimerCurrentlyRunning = true;
				if (exit)
				{
					exit = false;
					return;
				}
				
				if (cancelTimer)
				{
					cancelTimer = false;
					continue;
				}
				
				while (true)
				{
					try
					{
						long batchTimeRemaining = nextTimeToTimeoutBatch - System.currentTimeMillis();
						long waitTime = Math.min(batchTimeRemaining, maxTimeToWaitBetweenUpdates);
						wait(waitTime);
					} catch (InterruptedException e)
					{ }
					if (exit)
					{
						exit = false;
						return;
					}
					else if (cancelTimer)
					{
						batchTimerListener.cancelled();
						cancelTimer = false;
						break;
					}
					else
					{
						if (System.currentTimeMillis() >= nextTimeToTimeoutBatch)
						{
							batchTimerListener.timedOut(true);
							break;
						}
						else if (System.currentTimeMillis() >= nextTimeToTimeoutUpdates)
						{
							batchTimerListener.timedOut(false);
							break;
						}
					}
				}
			}
		}
	}
}
